package org.codehaus.activemq.message;

import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.ObjectStreamException;
import java.io.Serializable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.message.util.WireByteArrayInputStream;
import org.codehaus.activemq.message.util.WireByteArrayOutputStream;

public class DefaultWireFormat extends WireFormat
  implements Serializable
{
  private static final Log log = LogFactory.getLog(DefaultWireFormat.class);

  private final transient PacketReader messageReader = new ActiveMQMessageReader();
  private final transient PacketReader textMessageReader = new ActiveMQTextMessageReader();
  private final transient PacketReader objectMessageReader = new ActiveMQObjectMessageReader();
  private final transient PacketReader bytesMessageReader = new ActiveMQBytesMessageReader();
  private final transient PacketReader streamMessageReader = new ActiveMQStreamMessageReader();
  private final transient PacketReader mapMessageReader = new ActiveMQMapMessageReader();
  private final transient PacketReader messageAckReader = new MessageAckReader();
  private final transient PacketReader receiptReader = new ReceiptReader();
  private final transient PacketReader consumerInfoReader = new ConsumerInfoReader();
  private final transient PacketReader producerInfoReader = new ProducerInfoReader();
  private final transient PacketReader transactionInfoReader = new TransactionInfoReader();
  private final transient PacketReader xaTransactionInfoReader = new XATransactionInfoReader();
  private final transient PacketReader brokerInfoReader = new BrokerInfoReader();
  private final transient PacketReader connectionInfoReader = new ConnectionInfoReader();
  private final transient PacketReader sessionInfoReader = new SessionInfoReader();
  private final transient PacketReader durableUnsubscribeReader = new DurableUnsubscribeReader();
  private final transient PacketReader reponseReceiptReader = new ResponseReceiptReader();
  private final transient PacketReader intReponseReceiptReader = new IntResponseReceiptReader();
  private final transient PacketReader capacityInfoReader = new CapacityInfoReader();
  private final transient PacketReader capacityInfoRequestReader = new CapacityInfoRequestReader();

  private final transient PacketWriter messageWriter = new ActiveMQMessageWriter();
  private final transient PacketWriter textMessageWriter = new ActiveMQTextMessageWriter();
  private final transient PacketWriter objectMessageWriter = new ActiveMQObjectMessageWriter();
  private final transient PacketWriter bytesMessageWriter = new ActiveMQBytesMessageWriter();
  private final transient PacketWriter streamMessageWriter = new ActiveMQStreamMessageWriter();
  private final transient PacketWriter mapMessageWriter = new ActiveMQMapMessageWriter();
  private final transient PacketWriter messageAckWriter = new MessageAckWriter();
  private final transient PacketWriter receiptWriter = new ReceiptWriter();
  private final transient PacketWriter consumerInfoWriter = new ConsumerInfoWriter();
  private final transient PacketWriter producerInfoWriter = new ProducerInfoWriter();
  private final transient PacketWriter transactionInfoWriter = new TransactionInfoWriter();
  private final transient PacketWriter xaTransactionInfoWriter = new XATransactionInfoWriter();
  private final transient PacketWriter brokerInfoWriter = new BrokerInfoWriter();
  private final transient PacketWriter connectionInfoWriter = new ConnectionInfoWriter();
  private final transient PacketWriter sessionInfoWriter = new SessionInfoWriter();
  private final transient PacketWriter durableUnsubscribeWriter = new DurableUnsubscribeWriter();
  private final transient PacketWriter reponseReceiptWriter = new ResponseReceiptWriter();
  private final transient PacketWriter intReponseReceiptWriter = new IntResponseReceiptWriter();
  private final transient PacketWriter capacityInfoWriter = new CapacityInfoWriter();
  private final transient PacketWriter capacityInfoRequestWriter = new CapacityInfoRequestWriter();
  private transient WireByteArrayOutputStream internalBytesOut;
  private transient DataOutputStream internalDataOut;
  private transient WireByteArrayInputStream internalBytesIn;
  private transient DataInputStream internalDataIn;

  public DefaultWireFormat()
  {
    this.internalBytesOut = new WireByteArrayOutputStream();
    this.internalDataOut = new DataOutputStream(this.internalBytesOut);
    this.internalBytesIn = new WireByteArrayInputStream();
    this.internalDataIn = new DataInputStream(this.internalBytesIn);
  }

  public WireFormat copy() {
    return new DefaultWireFormat();
  }

  public Packet readPacket(DataInput in) throws IOException {
    int type = in.readByte();
    return readPacket(type, in);
  }

  public Packet readPacket(int firstByte, DataInput dataIn) throws IOException {
    switch (firstByte) {
    case 6:
      return readPacket(dataIn, this.messageReader);
    case 7:
      return readPacket(dataIn, this.textMessageReader);
    case 8:
      return readPacket(dataIn, this.objectMessageReader);
    case 9:
      return readPacket(dataIn, this.bytesMessageReader);
    case 10:
      return readPacket(dataIn, this.streamMessageReader);
    case 11:
      return readPacket(dataIn, this.mapMessageReader);
    case 15:
      return readPacket(dataIn, this.messageAckReader);
    case 16:
      return readPacket(dataIn, this.receiptReader);
    case 17:
      return readPacket(dataIn, this.consumerInfoReader);
    case 18:
      return readPacket(dataIn, this.producerInfoReader);
    case 19:
      return readPacket(dataIn, this.transactionInfoReader);
    case 20:
      return readPacket(dataIn, this.xaTransactionInfoReader);
    case 21:
      return readPacket(dataIn, this.brokerInfoReader);
    case 22:
      return readPacket(dataIn, this.connectionInfoReader);
    case 23:
      return readPacket(dataIn, this.sessionInfoReader);
    case 24:
      return readPacket(dataIn, this.durableUnsubscribeReader);
    case 25:
      return readPacket(dataIn, this.reponseReceiptReader);
    case 26:
      return readPacket(dataIn, this.intReponseReceiptReader);
    case 27:
      return readPacket(dataIn, this.capacityInfoReader);
    case 28:
      return readPacket(dataIn, this.capacityInfoRequestReader);
    case 12:
    case 13:
    case 14:
    }log.error("Could not find PacketReader for packet type: " + AbstractPacket.getPacketTypeAsString(firstByte));
    return null;
  }

  public void writePacket(Packet packet, DataOutput dataOut)
    throws IOException
  {
    PacketWriter writer = getWriter(packet);
    if (writer != null)
      writePacket(packet, dataOut, writer);
  }

  public byte[] toBytes(Packet packet)
    throws IOException
  {
    byte[] data = null;
    PacketWriter writer = getWriter(packet);
    if (writer != null) {
      this.internalBytesOut.reset();
      this.internalDataOut.writeByte(packet.getPacketType());
      this.internalDataOut.writeInt(-1);
      writer.writePacket(packet, this.internalDataOut);
      this.internalDataOut.flush();
      data = this.internalBytesOut.toByteArray();

      int length = data.length - 5;
      packet.setMemoryUsage(length);

      data[1] = (byte)(length >>> 24 & 0xFF);
      data[2] = (byte)(length >>> 16 & 0xFF);
      data[3] = (byte)(length >>> 8 & 0xFF);
      data[4] = (byte)(length >>> 0 & 0xFF);
    }
    return data;
  }

  protected final synchronized void writePacket(Packet packet, DataOutput dataOut, PacketWriter writer) throws IOException
  {
    dataOut.writeByte(packet.getPacketType());
    this.internalBytesOut.reset();
    writer.writePacket(packet, this.internalDataOut);
    this.internalDataOut.flush();

    byte[] data = this.internalBytesOut.getData();
    int count = this.internalBytesOut.size();
    dataOut.writeInt(count);

    packet.setMemoryUsage(count);
    dataOut.write(data, 0, count);
  }

  protected final synchronized Packet readPacket(DataInput dataIn, PacketReader reader) throws IOException {
    Packet packet = reader.createPacket();
    int length = dataIn.readInt();
    packet.setMemoryUsage(length);

    byte[] data = new byte[length];
    dataIn.readFully(data);

    this.internalBytesIn.restart(data);
    reader.buildPacket(packet, this.internalDataIn);
    return packet;
  }

  private Object readResolve() throws ObjectStreamException {
    return new DefaultWireFormat();
  }

  private PacketWriter getWriter(Packet packet) throws IOException {
    PacketWriter answer = null;
    switch (packet.getPacketType()) {
    case 6:
      answer = this.messageWriter;
      break;
    case 7:
      answer = this.textMessageWriter;
      break;
    case 8:
      answer = this.objectMessageWriter;
      break;
    case 9:
      answer = this.bytesMessageWriter;
      break;
    case 10:
      answer = this.streamMessageWriter;
      break;
    case 11:
      answer = this.mapMessageWriter;
      break;
    case 15:
      answer = this.messageAckWriter;
      break;
    case 16:
      answer = this.receiptWriter;
      break;
    case 17:
      answer = this.consumerInfoWriter;
      break;
    case 18:
      answer = this.producerInfoWriter;
      break;
    case 19:
      answer = this.transactionInfoWriter;
      break;
    case 20:
      answer = this.xaTransactionInfoWriter;
      break;
    case 21:
      answer = this.brokerInfoWriter;
      break;
    case 22:
      answer = this.connectionInfoWriter;
      break;
    case 23:
      answer = this.sessionInfoWriter;
      break;
    case 24:
      answer = this.durableUnsubscribeWriter;
      break;
    case 25:
      answer = this.reponseReceiptWriter;
      break;
    case 26:
      answer = this.intReponseReceiptWriter;
      break;
    case 27:
      answer = this.capacityInfoWriter;
      break;
    case 28:
      answer = this.capacityInfoRequestWriter;
      break;
    case 12:
    case 13:
    case 14:
    default:
      log.error("no PacketWriter for packet: " + packet);
    }
    return answer;
  }
}